package com.wlz.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @author wlz
 * @date 2022-07-05  10:08 下午
 */
@Slf4j
@Component
public class MyConsumer {


    @KafkaListener(topics = "kafka-spring-boot-topic",groupId = "consumer_group_1")
    public void listenerConsumerGroup1(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        log.info("listenerConsumerGroup1 value = {}", value);
        log.info("listenerConsumerGroup1 record = {}", record);

        //手动提交offset
        //ack.acknowledge();

    }

    @KafkaListener(topics = "kafka-spring-boot-topic",groupId = "consumer_group_2")
    public void listenerConsumerGroup2(ConsumerRecord<String,String> record, Acknowledgment ack) {
        String value = record.value();
        log.info("listenerConsumerGroup2 value = {}", value);
        log.info("listenerConsumerGroup2 record = {}", record);

        //手动提交offset
        //ack.acknowledge();

    }

}

